package sample.persistence

//#persistent-actor-example
import akka.actor._
import akka.persistence._

case class Cmd(data: String)
case class Evt(data: String)

case class ExampleState(events: List[String] = Nil) {
  def updated(evt: Evt): ExampleState = copy(evt.data :: events)
  def size: Int = events.length
  override def toString: String = events.reverse.toString
}

class ExamplePersistentActor extends PersistentActor {
  override def persistenceId = "sample-id-1"

  //skip loading snapshots, may be its current serialization format has changed in a incompatible way
  //override def recovery: Recovery = Recovery(fromSnapshot = SnapshotSelectionCriteria.None)

  //disable recovery
  //override def recovery: Recovery = Recovery.none

  var state = ExampleState()

  def updateState(event: Evt): Unit =
    state = state.updated(event)

  def numEvents =
    state.size

  val receiveRecover: Receive = {
    case evt: Evt =>
      println("recovering ",evt.data)
      updateState(evt)
    case SnapshotOffer(_, snapshot: ExampleState) =>
      println("recv snapshotoffer")
      state = snapshot
    case RecoveryCompleted =>
      //在恢复之后，和处理新消息之前，会收到这个
      println("recovery completed")
  }

  override protected def onRecoveryFailure(cause: Throwable, event: Option[Any]): Unit = {
    println("recovery failed")
    super.onRecoveryFailure(cause,event)
  }

  val receiveCommand: Receive = {
    case Cmd(data) =>
      persist(Evt(s"${data}-${numEvents}"))(updateState)
      persist(Evt(s"${data}-${numEvents + 1}")) { event =>
        updateState(event)
        context.system.eventStream.publish(event)
      }
    case "snap"  =>
      println("saving snapshot...")
      saveSnapshot(state)
    case "print" => println(state)
  }

}
//#persistent-actor-example

object PersistentActorExample extends App {

  val system = ActorSystem("example")
  val persistentActor = system.actorOf(Props[ExamplePersistentActor], "persistentActor-4-scala")

  persistentActor ! Cmd("FOO")
  persistentActor ! Cmd("BAZ")
  persistentActor ! Cmd("BAR")
  persistentActor ! "snap"
  persistentActor ! Cmd("BUZZ")
  persistentActor ! "print"

  Thread.sleep(10000)
  system.terminate()
}
